#!/usr/bin/env python
# -*- coding: UTF-8 -*-
'''
@Project ：henan_sumo
@File ：main_send.py
@IDE ：PyCharm
@Author ：xinyingjie
@Date ：2025/4/16 17:48
'''
import threading
from copy import copy
from multiprocessing import Queue, Process
from loguru import logger
import time
import json
from kafka import KafkaConsumer, KafkaProducer

from config.config import KAFKA_HOST
from utils.config_temp import RADAR_DATA_LIST_CJ, RADAR_DATA_LIST_JJ, first_CJ, first_JJ
from utils.tools import time_str


class DataFusion:
    def __init__(self, receive_topic, org_code_list, send_radar_topic, send_kako_topic, RADAR_DATA_LIST, first_sn):
        self.RADAR_DATA_LIST = RADAR_DATA_LIST
        self.first_sn_list = first_sn
        self.receive_topic = receive_topic
        self.org_code_list = org_code_list
        self.send_radar_topic = send_radar_topic
        self.send_kako_topic = send_kako_topic
        self.topic_dict = {}
        self.plate_sn_dict = {}
        self.kako_cache_dict = {}
        self.init_kafka()

    def init_kafka(self):
        for sn in self.RADAR_DATA_LIST.keys():
            self.topic_dict[sn] = None
            self.kako_cache_dict[sn] = {}
            self.plate_sn_dict[sn] = {}

        threading.Thread(target=self.receive_data, args=(receive_topic,)).start()

    def start_send(self):
        threading.Thread(target=self.send_radar_kafka).start()
        threading.Thread(target=self.send_kako_kafka).start()

    def receive_data(self, kafka_topic):
        kafka_consumer_e1 = KafkaConsumer(
            kafka_topic,
            bootstrap_servers=KAFKA_HOST,
            auto_offset_reset='latest',
            enable_auto_commit=False
        )
        while True:
            time.sleep(0.1)
            target_data = []
            target_result = {}
            for message in kafka_consumer_e1:
                if message:
                    try:
                        message_value = message.value.decode('utf-8')
                        E1_info = json.loads(message_value)
                        if 'targetList' in E1_info.keys():
                            orgCode = E1_info['orgCode'].split("-")[-1]
                            self.topic_dict[orgCode] = E1_info
                    except Exception as e:
                        logger.info(f"kafka数据处理异常{e}")

    def send_radar_kafka(self):
        kafka_producer = KafkaProducer(bootstrap_servers=KAFKA_HOST, key_serializer=str.encode,
                                       value_serializer=lambda x: json.dumps(x).encode('utf-8'))
        frame_num = 0
        while True:
            all_list = []
            for key, e1_value in self.topic_dict.items():
                if e1_value:
                    if 'targetList' not in e1_value.keys():
                        continue
                    targetList = e1_value['targetList']
                    for target in targetList:
                        target_c = target.copy()
                        target_c['id'] = str(target_c['id']) + "-" + str(target_c['station'])
                        all_list.append(target_c)
                    # all_list += e1_value['targetList']
            send_data = {"frameNum": frame_num, 'globalTime': time_str(int(time.time() * 1000)), "targetList": all_list}
            # print(send_data)
            kafka_producer.send(self.send_radar_topic, value=send_data,
                                key=str(int(time.time() * 1000)))
            frame_num += 1
            time.sleep(0.1)

    def send_kako_kafka(self):
        kafka_producer = KafkaProducer(bootstrap_servers=KAFKA_HOST, key_serializer=str.encode,
                                       value_serializer=lambda x: json.dumps(x).encode('utf-8'))
        while True:
            for key, e1_value in self.topic_dict.items():
                if e1_value:
                    if 'targetList' not in e1_value.keys():
                        continue
                    target_list = e1_value['targetList']
                    for target in target_list:
                        car_id = target['id']
                        station = target['station']
                        if station in self.first_sn_list:
                            if car_id not in self.kako_cache_dict[station].keys():
                                target['timestamp'] = time_str(int(time.time() * 1000))
                                target['global_id'] = str(car_id) + "-" + str(station)
                                picLicense = target['picLicense']
                                if "默A" not in picLicense:
                                    self.plate_sn_dict[station][picLicense] = [int(time.time()), target['global_id']]
                                self.kako_cache_dict[station][car_id] = int(time.time())
                                logger.info(f"卡口上报{target}")
                                kafka_producer.send(self.send_kako_topic, value=target,
                                                    key=str(int(time.time() * 1000)))
                        else:
                            if car_id not in self.kako_cache_dict[station].keys():
                                picLicense = target['picLicense']
                                for first_sn in self.first_sn_list:
                                    if picLicense in self.plate_sn_dict[first_sn].keys():
                                        target['timestamp'] = time_str(int(time.time() * 1000))
                                        target['global_id'] = self.plate_sn_dict[first_sn][picLicense][1]
                                        logger.info(f"卡口上报{target}")
                                        kafka_producer.send(self.send_kako_topic, value=target,
                                                            key=str(int(time.time() * 1000)))
                                self.kako_cache_dict[station][car_id] = int(time.time())

            time.sleep(1)


if __name__ == '__main__':
    topic_list = ['CJ-10-SUM', 'JJ-10-SUM']
    org_code_list = ['CJ-10-SUM', 'JJ-10-SUM']
    send_radar_topic = "fusion_E1_data"
    send_kako_topic = "fusion_kako_data"
    receive_topic = "CJ-10-SUM"
    data_fusion = DataFusion(receive_topic, org_code_list, send_radar_topic, send_kako_topic, RADAR_DATA_LIST_CJ,
                             first_CJ)
    data_fusion.start_send()
